package com.ysy.serviceImpl;


import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.RandomUtil;
import com.alibaba.fastjson2.JSON;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ysy.constant.RedisConstant;
import com.ysy.constant.RocketMqConstant;
import com.ysy.context.SecurityContextHolder;
import com.ysy.dto.UserDTO;
import com.ysy.entity.Follow;
import com.ysy.entity.dto.FollowDTO;
import com.ysy.entity.req.FollowListReqDTO;
import com.ysy.entity.req.FollowReqDTO;
import com.ysy.entity.rsp.FollowListRspVO;
import com.ysy.enums.RedisLuaResultEnum;
import com.ysy.enums.ResponseCodeEnum;
import com.ysy.facade.UserFeignService;
import com.ysy.mapper.FollowMapper;

import com.ysy.persimmons.common.exception.ServiceException;
import com.ysy.persimmons.common.request.CursorPageBaseReq;
import com.ysy.persimmons.common.response.CursorPageBaseResp;
import com.ysy.persimmons.common.response.PageResponse;
import com.ysy.persimmons.common.response.Response;
import com.ysy.persimmons.common.util.AssertUtil;
import com.ysy.persimmons.common.util.DateUtils;
import com.ysy.service.FollowService;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.math.RandomUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.ClassPathResource;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scripting.support.ResourceScriptSource;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.time.LocalDateTime;
import java.util.*;

/**
 * <p>
 * 用户关注表 服务实现类
 * </p>
 *
 * @author yangSy
 * @since 2025-04-26
 */
@Slf4j
@Service
public class FollowServiceImpl extends ServiceImpl<FollowMapper, Follow> implements FollowService {



    @Resource
    private RedisTemplate<String,Object> redisTemplate;
    @Resource
    private RocketMQTemplate rocketMQTemplate;



    @Resource
    private UserFeignService userFeignService;

    @Resource
    private FollowMapper followMapper;
    @Autowired
    private ThreadPoolTaskExecutor threadPoolTaskExecutor;


    @Override
    public Response<?> followUser(FollowReqDTO followDTO) {

        // 参数是否正常
        Long followId = followDTO.getFollowId();
        AssertUtil.isFalse(followId==null || followId==0L, ResponseCodeEnum.PARAM_NOT_VALID);
        // 关注用户是不是自己
        Long userId = SecurityContextHolder.getUserId();
        AssertUtil.isFalse(userId==0L, ResponseCodeEnum.SYSTEM_ERROR);
        AssertUtil.isFalse(userId.equals(followId), ResponseCodeEnum.FOLLOW_IS_ONESELF);

        // 判断关注的用户是否存在
        UserDTO byId = userFeignService.findById(followId);
        AssertUtil.isFalse(Objects.isNull(byId), ResponseCodeEnum.USER_IS_NULL);

        // 是否已经关注过了,是否到达最大的限制数了
        DefaultRedisScript<Long> script1 = new DefaultRedisScript<>();
        ClassPathResource classPathResource = new ClassPathResource("/lua/z_set_follow_check_add.lua");
        script1.setScriptSource(new ResourceScriptSource(classPathResource));
        script1.setResultType(Long.class);

        String redisFollowKey = RedisConstant.buildUserFollowKey(followId);
        LocalDateTime localDateTimeNow = LocalDateTime.now();
        long time = DateUtils.localDateTimeTimestamp(localDateTimeNow);

        Long redisLuaResult = redisTemplate.execute(script1, Collections.singletonList(redisFollowKey), followId, time);
        log.info(" redisLua followUser Result:{}",redisLuaResult);
        AssertUtil.isFalse(Objects.isNull(redisLuaResult), ResponseCodeEnum.SYSTEM_ERROR);
        checkLuaScriptResult(redisLuaResult);

        RedisLuaResultEnum redisLuaResultEnum = RedisLuaResultEnum.getRedisLuaResultEnum(redisLuaResult);
        //   yield
        switch (redisLuaResultEnum) {
            case NOT_EXIST -> {
                // 不存在就要查询数据库批量添加添加
                QueryWrapper<Follow> queryWrapper = new QueryWrapper<>();
                queryWrapper.eq("user_id", userId);
                List<Follow> followList = followMapper.selectList(queryWrapper);

                int expireTime =60*60*5 + RandomUtils.nextInt(10);

                if (CollectionUtils.isEmpty(followList)) {
                    DefaultRedisScript<Long> Script2 = new DefaultRedisScript<>();
                    Script2.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/z_set_follow_add_expire.lua")));
                    Script2.setResultType(Long.class);
                    // 根据不同用户流量设置过期时间，设置 TODO
                    Long redisLuaEmptyResult = redisTemplate.execute(Script2, Collections.singletonList(redisFollowKey), followId, time,expireTime);
                    log.info(" redisLua followUser redisLuaEmpty Result:{}",redisLuaEmptyResult);
                } else {
                    int argsLength = followList.size() * 2 + 1;
                    Object[] luaArgs = new Object[argsLength];
                    int i = 0;
                    for (Follow follow : followList) {
                        // 关注时间 score
                        luaArgs[i] = DateUtils.localDateTimeTimestamp(follow.getCreateTime());
                        // 关注的用户 ID ZSet value
                        luaArgs[i + 1] = follow.getFollowUserId();
                        i += 2;
                    }
                    //  ZSet 的过期时间
                    luaArgs[argsLength - 1] = expireTime;
                    // 批量写入缓存
                    DefaultRedisScript<Long> script3 = new DefaultRedisScript<>();
                    script3.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/z_set_follow_batch_add_expire.lua")));
                    script3.setResultType(Long.class);
                    redisTemplate.execute(script3, Collections.singletonList(redisFollowKey), luaArgs);

                    // 调用上面的 Lua 脚本：z_set_follow_check_add.lua , 添加当前新加入的关注
                    redisLuaResult = redisTemplate.execute(script1, Collections.singletonList(redisFollowKey), followId, time);
                    checkLuaScriptResult(redisLuaResult);

                }
            }
            case NOT_FOLLOW -> {
                // 正常这步是走不到上面会加入 未关注,执行脚本关注
            }
        }

        //MQ 异步写入数据库，需要注意强一致性需要存根
        FollowDTO followMq = new FollowDTO();
        followMq.setUserId(userId);
        followMq.setFollowUserId(followId);
        followMq.setCreateTime(localDateTimeNow);
        log.info("==> 准备MQ发送关注操作消息体: {}", followMq);
        MessageBuilder<String> followDTOMessageBuilder = MessageBuilder.withPayload(JSON.toJSONString(followMq));
        String topic = RocketMqConstant.TOPIC_FOLLOW_CANCEL +":"+ RocketMqConstant.FOLLOW;
        // 同一组业务相关的消息必须被发送到同一个队列,Orderly表示顺序发送
        rocketMQTemplate.asyncSendOrderly(topic,followDTOMessageBuilder,userId.toString(),new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("==> MQ发送关注操作onSuccess: {}", sendResult);
                // 注意补偿机制或者存根
            }
            @Override
            public void onException(Throwable throwable) {
                log.info("==> MQ发送关注操作onException:", throwable);
            }
        });

        return Response.success();
    }

    /**
     * 检查是否已经添加，未添加就加入
     * @param redisLuaResult
     */
    private void checkLuaScriptResult(Long redisLuaResult) {
        RedisLuaResultEnum redisLuaResultEnum = RedisLuaResultEnum.getRedisLuaResultEnum(redisLuaResult);
        AssertUtil.isFalse(Objects.isNull(redisLuaResultEnum), ResponseCodeEnum.SYSTEM_ERROR);
        switch (redisLuaResultEnum) {
            case ALREADY_FOLLOW -> throw new ServiceException(ResponseCodeEnum.FOLLOW_ALREADY_USER);
            case FOLLOW_LIMIT -> throw new ServiceException(ResponseCodeEnum.FOLLOW_USER_LIMIT_MAX);
        }
    }

    /**
     * 害怕用户多次点击可以及一个用户key和关注用户的锁
     * @param followDTO
     * @return
     */
    @Override
    public Response<?> cancelFollow(FollowReqDTO followDTO) {

        // 参数是否正常
        Long followId = followDTO.getFollowId();
        AssertUtil.isFalse(followId==null || followId==0L, ResponseCodeEnum.PARAM_NOT_VALID);

        // 取消关注用户是不是自己
        Long userId = SecurityContextHolder.getUserId();
        AssertUtil.isFalse(userId==0L, ResponseCodeEnum.SYSTEM_ERROR);
        AssertUtil.isFalse(userId.equals(followId), ResponseCodeEnum.CANCEL_IS_ONESELF);

        // 判断取关的用户是否存在
        UserDTO byId = userFeignService.findById(followId);
        AssertUtil.isFalse(Objects.isNull(byId), ResponseCodeEnum.USER_IS_NULL);

        // 查询redis之前是否关注过该用户
        // redis存在用户关注信息，进行删除，但是数据库其他接口可能也再更新，数据库兜底
        DefaultRedisScript<Long> script1 = new DefaultRedisScript<>();
        script1.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/z_set_cancel_follow.lua")));
        script1.setResultType(Long.class);

        String userFollowKey = RedisConstant.buildUserFollowKey(userId);
        Long  cancelResult = redisTemplate.execute(script1,Collections.singletonList(userFollowKey),followId);

        RedisLuaResultEnum redisLuaResultEnum = RedisLuaResultEnum.getRedisLuaResultEnum(cancelResult);
        AssertUtil.isFalse(Objects.isNull(redisLuaResultEnum), ResponseCodeEnum.SYSTEM_ERROR);
        switch (redisLuaResultEnum) {
            case NOT_FOLLOW -> throw new ServiceException(ResponseCodeEnum.IS_NO_FOLLOW);
            case NOT_EXIST -> {
                // 1、查询不到redis，查询数据库更新缓存
                List<Follow> userFollowList = followMapper.selectList(new QueryWrapper<Follow>().eq("user_id", userId));
                if (CollectionUtils.isEmpty(userFollowList)) {
                    //1.1 查询没有数据，结束
                    return Response.fail("您未关注该用户");
                } else {
                    // 批量跟新缓存
                    int exTime = 60*60*5 + RandomUtil.randomInt(10);
                    Object[] args = buildLuaArgs(userFollowList,exTime);
                    DefaultRedisScript<Long> script2 = new DefaultRedisScript<>();
                    script2.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/z_set_cancel_follow.lua")));
                    script2.setResultType(Long.class);
                    redisTemplate.execute(script2,Collections.singletonList(userFollowKey),args);

                    // 这里再删除，并取关
                    cancelResult = redisTemplate.execute(script1,Collections.singletonList(userFollowKey),followId);
                    log.info("查询数据库之后更新缓存执行取关结果{}-{}：{}", userId, followId, cancelResult);
                    if (RedisLuaResultEnum.getRedisLuaResultEnum(cancelResult)==RedisLuaResultEnum.NOT_FOLLOW){
                        throw new ServiceException(ResponseCodeEnum.IS_NO_FOLLOW);
                    }
                }
            }

        }
        // 异步取执行更新数据库
        Follow followCancel = new Follow();
        followCancel.setUserId(userId);
        followCancel.setFollowUserId(followId);
        MessageBuilder<String> messageBuilder = MessageBuilder.withPayload(JSON.toJSONString(followCancel));
        String topic = RocketMqConstant.TOPIC_FOLLOW_CANCEL +":"+ RocketMqConstant.CANCEL;
        rocketMQTemplate.asyncSendOrderly(topic,messageBuilder,userId.toString(),new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                log.info("onSuccess查执行取关结果{}-{}:{}", userId, followId, sendResult);
            }
            @Override
            public void onException(Throwable throwable) {
                log.info("执行取关结果{}-{}onException:", userId, followId, throwable);
            }
        });

        return Response.success();
    }

    @Override
    public PageResponse<FollowListRspVO> followList(FollowListReqDTO follow) {

        // 查询是否允许他人查看关注 TODO
        // 查询zset中的范围数据根据score排序
        String fansKey = RedisConstant.buildUserFansKey(follow.getUserId());
        long total = redisTemplate.opsForZSet().zCard(fansKey);
        Integer pageSize =  10;
        Integer pageNo = follow.getPageNo();
        List<FollowListRspVO> followListRspVOS = null;
         if (total > 0) {
             // 计算一共多少页
             long totalPage = PageResponse.getTotalPage(total, pageSize);
             // 请求的页码超出了总页数
             if (pageNo > totalPage) return PageResponse.success(null, pageNo, total);
             // 准备从 Redis 中查询 ZSet 分页数据
             // 每页 10 个元素，计算偏移量
             long offset = PageResponse.getOffset(pageNo, pageSize);
             Set<Object> objects = redisTemplate.opsForZSet().reverseRangeByScore(fansKey, Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, offset, pageSize);
             if (CollUtil.isNotEmpty(objects)) {
                 List<Long> userIds = objects.stream().map(object -> Long.valueOf(object.toString())).toList();
                 // 查询用户信息
                 extracted(userIds, followListRspVOS);
             }
         }else {

             // 若 Redis 中没有数据，则从数据库查询
             long count = followMapper.selectCount(new QueryWrapper<Follow>().eq("user_id", follow.getUserId()));
             long totalPage = PageResponse.getTotalPage(count, pageSize);
             if (pageNo > totalPage) {
                 return PageResponse.success(null, pageNo, count);
             };
             long offset = PageResponse.getOffset(pageNo, pageSize);
             // 分页查询
             List<Follow> followList = followMapper.selectPageListByUserId(follow.getUserId(), offset, pageSize);
             if (CollUtil.isNotEmpty(followList)) {
                 // 提取所有用户 ID 到集合中
                 List<Long> userIds = followList.stream().map(Follow::getFollowUserId).toList();
                 // 查询用户信息
                 extracted(userIds, followListRspVOS);
                 //同步到redis中
                 threadPoolTaskExecutor.execute(()->{
                     asyncRedisCache(follow.getUserId());
                 });
             }
         }
        return PageResponse.success(followListRspVOS, pageNo, total);
    }

    /**
     * 缓存redis中的关注数量
     * @param userId
     */
    private void asyncRedisCache(Long userId) {

        List<Follow> followList = followMapper.selectByIds(List.of(userId));


        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>();
        redisScript.setScriptSource(new ResourceScriptSource(new ClassPathResource("/lua/z_set_follow_batch_add_expire.lua")));
        redisScript.setResultType(Long.class);

        String redisFollowKey = RedisConstant.buildUserFollowKey(userId);
        // 5小时+随机秒数
        long expireSeconds = 60*60*5 + RandomUtil.randomInt(10);
        // 构建 Lua 参数
        Object[] luaArgs = buildLuaArgs(followList, expireSeconds);
        redisTemplate.execute(redisScript,Collections.singletonList(redisFollowKey),luaArgs);


    }

    /**
     * 构建lua脚本参数
     * @param followList
     * @param exTime
     * @return
     */
    private Object[] buildLuaArgs(List<Follow> followList, long exTime) {
        int size = followList.size() +1 ;
        Object[] args = new Object[size];
        int i = 0;
        for (Follow follow : followList) {
            args[i] = DateUtils.localDateTimeTimestamp(follow.getCreateTime());
            args[i+1] = follow.getFollowUserId();
            i+=2;
        }
        args[size-1] = exTime;
        return args;
    }

    private void extracted(List<Long> userIds, List<FollowListRspVO> followListRspVOS) {
        List<UserDTO> userDTOList = userFeignService.findByIds(userIds);

        userDTOList.forEach(userDTO -> {
            FollowListRspVO followListRspVO = new FollowListRspVO();
            followListRspVO.setAvatar(userDTO.getAvatar());
            followListRspVO.setName(userDTO.getName());
            followListRspVO.setIntroduction(userDTO.getIntroduction());
            followListRspVO.setUserId(userDTO.getId());
            followListRspVOS.add(followListRspVO);
        });
    }


//    @Override
//    public CursorPageBaseResp<FollowListRspVO> followList(CursorPageBaseReq follow) {
//       if (follow.isFirstPage()) {
//           follow.setCursor("0");
//       }
//        // 判断当前关注是否仅自己可见 TODO
//
//        // 查询zset中的范围数据根据score排序
//        String fansKey = RedisConstant.buildUserFansKey(follow.getUserId());
//        long total = redisTemplate.opsForZSet().zCard(fansKey);
//
//        Integer pageSize = follow.getPageSize();
//        long cursor = Long.parseLong(follow.getCursor());
//
//        List<FollowListRspVO> followListRspVOS = null;
//        long size =  cursor+Long.valueOf(pageSize);
//
//
//        if (total > 0) {
//            // 当前查询大于最大说明已经到底了，在查询只能去数据库，先不做
//            if (cursor>total) {
//                return CursorPageBaseResp.empty();
//            }
//
//            // 从redis中获取
//            // Double.NEGATIVE_INFINITY 和 Double.POSITIVE_INFINITY 的作用
//            // min = Double.NEGATIVE_INFINITY：表示 分数的最小值无下限（即包含所有分数 ≤ max 的成员）。
//            // max = Double.POSITIVE_INFINITY：表示 分数的最大值无上限（即包含所有分数 ≥ min 的成员）。
//            // 如果同时使用 NEGATIVE_INFINITY 和 POSITIVE_INFINITY，相当于 不限制分数范围，返回所有成员（但仍受 offset 和 limit 限制）
//            Set<Object> objects = redisTemplate.opsForZSet().reverseRangeByScore(fansKey, Double.NEGATIVE_INFINITY, Double.POSITIVE_INFINITY, cursor, cursor + pageSize);
//
//            if (CollUtil.isNotEmpty(objects)) {
//                // 提取所有用户 ID 到集合中
//                List<Long> userIds = objects.stream().map(object -> Long.valueOf(object.toString())).toList();
//                // 查询用户信息
//                List<UserDTO> userDTOList = userFeignService.findByIds(userIds);
//
//                userDTOList.forEach(userDTO -> {
//                    FollowListRspVO followListRspVO = new FollowListRspVO();
//                    followListRspVO.setAvatar(userDTO.getAvatar());
//                    followListRspVO.setName(userDTO.getName());
//                    followListRspVO.setIntroduction(userDTO.getIntroduction());
//                    followListRspVO.setUserId(userDTO.getId());
//                    followListRspVOS.add(followListRspVO);
//                });
//            }
//        }else {
//            // 从数据库获取
//            // 先查询记录总量
//
//            long count = followMapper.selectCount(new QueryWrapper<Follow>().eq("user_id", follow.getUserId()));
//            if (cursor>count) {
//                return CursorPageBaseResp.empty();
//            }
//
//            // 偏移量 id 去掉了导致
//            List<Follow> followList =  followMapper.selectCursorList(follow.getUserId(),cursor,follow.getPageSize());
//
//
//
//
//        }
//        CursorPageBaseResp cursorPageBaseResp =new CursorPageBaseResp();
//        cursorPageBaseResp.setCursor(String.valueOf(cursor));
//        cursorPageBaseResp.setIsLast(size >= total);
//       return CursorPageBaseResp.init(cursorPageBaseResp,followListRspVOS) ;
//    }


}
